package com.gene.tlonpo.config;

import com.gene.tlonpo.exception.TlonpoException;
import com.gene.tlonpo.listener.EventMqListener;
import com.gene.tlonpo.service.remote.server.TlonpoNettyServer;
import com.gene.tlonpo.util.AssertUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

/**
 * @author geneX
 * 组件初始化器
 */
@Slf4j
public class TlonpoInitializer {

    /**
     * 不可重复初始化
     */
    private static volatile TlonpoInitializer tlonpoInitializer;

    private TlonPoInitConfig config;

    public synchronized static TlonpoInitializer buildInitizlizer(TlonPoInitConfig config) {
        if (tlonpoInitializer != null) {
            throw new TlonpoException("a initizlizer has exist");
        }

        tlonpoInitializer = new TlonpoInitializer(config);
        return tlonpoInitializer;
    }

    protected TlonpoInitializer(TlonPoInitConfig config) {
        AssertUtil.isNotNull(config);
        config.check();
        this.config = config;
    }


    public void init() {

        //内置服务端启动
        startNettyServer();

        //消息监听启动
        startRocketMqListener();

        //初始化静态上下文
        initStaticContext();
    }

    protected void initStaticContext() {
        //实例化内置redis客户端
        LonpoStaticContext.redisConfig = newRedisConfig();
        LonpoStaticContext.port = this.config.getPort();
        LonpoStaticContext.globalBizKey = this.config.getGlobalBizKey();
        LonpoStaticContext.globalTimeOut = this.config.getGlobalTimeOut();
        LonpoStaticContext.globalTimeOutUnit = this.config.getGlobalTimeOutUnit();
    }

    protected void startRocketMqListener() {
        try {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
            consumer.setNamesrvAddr(this.config.getConsumerProperties().getNameServer());
            consumer.setConsumerGroup(this.config.getConsumerProperties().getConsumerGroup());
            consumer.subscribe(this.config.getConsumerProperties().getTopic(), this.config.getConsumerProperties().getTag());
            consumer.setMessageModel(MessageModel.CLUSTERING);
            consumer.registerMessageListener(new EventMqListener());
            consumer.start();
        } catch (Exception e) {
            log.error("mqListener start fail", e);
            throw new TlonpoException("mqListener start fail");
        }

    }

    protected TlonpoNettyServer startNettyServer() {
        TlonpoNettyServer tlonpoNettyServer = new TlonpoNettyServer(this.config.getPort());
        tlonpoNettyServer.start();
        return tlonpoNettyServer;
    }

    private RedisConfig newRedisConfig() {
        return new RedisConfig(this.config.getRedisTemplate());
    }
}
